package distributedcounter;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;

/**
 * @author Yinzi Chen
 * @date Jan 27, 2014
 */
public class Distributor {

    // private static Map<String, PrintWriter> descriptorCache;
    private static String fileName;
    private static int machineNum;
    private static int diskNum;

    public static void main(String[] args) throws IOException {
        fileName = "/0/data/foo.txt";
        machineNum = 1000;
        diskNum = 12;
        if (args.length > 0) {
            fileName = args[0];
        }
        if (args.length > 1) {
            machineNum = Integer.parseInt(args[1]);
        }
        if (args.length > 2) {
            diskNum = Integer.parseInt(args[2]);
        }
        // descriptorCache = new HashMap<String, PrintWriter>();
        distributeText();
    }

    private static void distributeText() throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(
                new FileInputStream(fileName)));
        String line;
        int machineId, diskId;
        String destFile;
        while ((line = br.readLine()) != null) {
            machineId = getMachineId(line);
            diskId = getDiskId(line);
            destFile = "/0/data/" + machineId + "/" + diskId + "/data/foo.txt";
            PrintWriter pw = new PrintWriter(new OutputStreamWriter(
                    new FileOutputStream(destFile, true)));
            // cache the file descriptor
            // PrintWriter pw = descriptorCache.get(destFile);
            // if (pw == null) {
            // pw = new PrintWriter(new OutputStreamWriter(
            // new FileOutputStream(destFile, true)));
            // descriptorCache.put(destFile, pw);
            // }
            // if too many file descriptor, randomly remove one
            // if (descriptorCache.size() > 512) {
            // descriptorCache.
            // }
            pw.println(line);
            pw.close();
        }
        br.close();
    }

    // mapping to machine id
    private static int getMachineId(String text) {
        return Math.abs(text.hashCode()) % machineNum;
    }

    // mapping to disk id
    private static int getDiskId(String text) {
        int val = 0;
        for (int i = 0; i < text.length(); ++i) {
            val *= 10;
            val += text.charAt(i);
        }
        return Math.abs(val) % diskNum;
    }
}
